package sparkCore

import org.apache.spark.{Partition, SparkConf, SparkContext}

object Demo02_createAPI {

  /**
   * 1.可以读取单个文件，也可以读取一个目录下的所有文件
   *   目录下的所有文件以一个文件的形式表示
   * 2.并行度生效顺序：
   *   1.textFile 第二个参数
   *   2."spark.default.parallelism" 配置的值  但是需要与2做最小值判断
   *   3."spark.default.parallelism"为空，
   *      本地模式：当前节点的所有cpu核心数 但是需要与2做最小值判断
   *      独立模式或分布式模式：集群中所有节点的所有ＣＰＵ的核心数　但是需要与2做最小值判断
   * 3.分区策略
   *   与 hadoop 的文本的默认分区策略是一致的
   * @param sc
   */
  def useTextFile(sc:SparkContext) = {

    val lineRDD = sc.textFile("D:/data/data_spark/wordCount/")

    println(lineRDD.getNumPartitions)
    println(lineRDD.partitions.length)

    lineRDD.foreach(println)

  }

  /**
   * 1.可以读取单个文件，也可以读取整个目录下的所有文件
   *   读取目录下的多个文件时，是以二元组的形式表示，key是文件的路径，value是此文件的内容
   * 2.并行度：与textFile完全一致
   * 3.分区策略
   *   使用的是 MR 中的 CombineFileInputFormat 的分区策略
   * @param sc
   */
  def useWholeTextFiles(sc:SparkContext) = {
    val lineRDD = sc.wholeTextFiles("D:/data/data_spark/wordCount/")

    lineRDD.foreach(println)
  }

  /**
   * 1.使用已知的集合创建RDD
   * 2.并行度
   *   1.parallelize 第二个参数
   *   2."spark.default.parallelism" 配置的值
   *   3."spark.default.parallelism"为空，
   *     本地模式：当前节点的所有cpu核心数
   *     独立模式或分布式模式：集群中所有节点的所有ＣＰＵ的核心数
   * 3.分区策略 按照集合的元素数量根据源码的算法均分成多个分区
   * @param sc
   */
  def useParallelize(sc:SparkContext) = {
    val list = List(1,2,3,4,5,6,7,8,9,10)
    val rdd = sc.parallelize(list,3)

//    println(rdd.getNumPartitions)

    rdd.collect.foreach(println)

  }

  /**
   * 1.如果参数是普通的Seq序列，完全等价于 parallelize
   * 2.参数是 Seq[(T, Seq[String])] 类型的，忽略
   * @param sc
   * @return
   */
  def useMakeRDD(sc:SparkContext) = {
    val list = List(1,2,3,4,5,6,7,8,9,10)
    sc.makeRDD(list)
  }

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("create")
//    conf.set("spark.default.parallelism","6")
    val sc = new SparkContext(conf)

    useParallelize(sc)

//    Thread.sleep(1000000000)

  }

}
